package main

import (
	"database/sql"
	"time"

	"github.com/bwmarrin/snowflake"

	m "gitee.com/sting/ksync/messages"
	_ "github.com/lib/pq"
)

type Database struct {
	driver string
	url    string
}

var (
	db   *sql.DB
	node *snowflake.Node
)

func (dbs *Database) open() {
	var err error
	node, err = snowflake.NewNode(1)
	if err != nil {
		Log.Error(err)
		return
	}

	db, err = sql.Open(dbs.driver, dbs.url)
	if err != nil {
		Log.Error("error connecting to the database: ", err)
	}

	db.SetMaxOpenConns(2000)
	db.Ping()

	go func() {
		for range time.Tick(5 * time.Second) {
			Log.Debugf("db connection count stats: %v ", db.Stats())
		}
	}()

	// Create the "kafka event" table.
	/*
		type DBEventLog struct{
			Schema string `json:"schema"`
			Table string `json:"table"`
			Type string `json:"type"`
			Source string `json:"source"`
			EventId string `json:"eventId"`
			EventTime int `json:"eventTime"`
			Body string `json:"body"`
		}
	*/

	if _, err := db.Exec(
		`CREATE TABLE IF NOT EXISTS tm_event (
			id int PRIMARY KEY, 
			type STRING,
			name STRING,
			rec_id STRING,
			event_time  TIMESTAMPTZ,
			body STRING
			)`); err != nil {
		Log.Fatal(err)
	}
}

func (dbs *Database) saveEvent(evt *m.DBEventLog) error {
	if evt.EventId == "" {
		return nil
	}
	id := node.Generate()
	_, err := db.Exec(
		"INSERT INTO tm_event VALUES ($1,$2,$3,$4,$5,$6)",
		id,
		evt.Type,
		evt.Table,
		evt.EventId,
		time.Unix(int64(evt.EventTime), 0),
		evt.Body,
	)
	if err != nil {
		return err
	}
	Log.Debugf("write into database: [%s]", id)
	return nil
}
